#include "../common/config.h"
#include "fsorder.h"
LoggerPtr gLogger = Logger::getLogger("fsorder");
int  gRunThread = 1;


apr_queue_t* jobqueue = NULL;
apr_queue_t* resultqueue = NULL;

apr_queue_t* authqueue = NULL;
apr_queue_t* authresultqueue = NULL;



apr_queue_t* conn_queue = NULL;

void * APR_THREAD_FUNC fetch_order(apr_thread_t *thd, void *data);
void * APR_THREAD_FUNC result_order(apr_thread_t *thd, void *data);
void * APR_THREAD_FUNC fetch_auth(apr_thread_t *thd, void *data);
void * APR_THREAD_FUNC result_auth(apr_thread_t *thd, void *data);

void outcall(esl_handle_t* handle, ordercall_info_t* pJob)
{

    string strProtocol;
    strProtocol = "sofia/external/";
	
    vector<call_param_t> callParams;
	vector<call_param_t> callParamsb;
    if (!getCallParam(pJob->acct, pJob->caller, pJob->called, CALLBACK1, callParams))
    {
        //db failed, retry job
        //apr_queue_push (jobqueue, pJob);
		LOG4CXX_INFO(gLogger,"outcall getCallParam error:"<< pJob->acct << " caller:" << pJob->caller << " called:" << pJob->called);
        return;
    }

    if (callParams.begin()->nTimeout <= 0)
    {
		LOG4CXX_INFO(gLogger,"outcall nTimeout error:"<< pJob->acct << " caller:" << pJob->caller << " called:" << pJob->called << " time:" << callParams.begin()->nTimeout);
        return;
    }

	
    //设置拨号字符串
    stringstream ss;
	if (pJob->called == "empty")
	{
		ss << "bgapi originate {ignore_early_media=true,proxy_media=true,jitterbuffer_msec=180,call_timeout=50,orig_acct=" << pJob->acct << ",orig_caller=" << pJob->caller << ",orig_called=" << pJob->called << "}";
		int n = 0;

		for (vector<call_param_t>::iterator it = callParams.begin(); it != callParams.end(); ++it)
		{
			if (n > 0)
			{
				ss << "|";
			}

			ss << "[calltype=1,execute_on_answer=\'sched_hangup +" << it->nTimeout << " alloted_timeout\',origination_caller_id_name=\'" << it->caller << "\',origination_caller_id_number=" << it->caller << ",absolute_codec_string=\'" << it->codec << "\']"
			<< strProtocol << it->called << "@" << it->server;
			n++;
		}

		ss << " " << pJob->called << " XML callback " << pJob->caller << " " << pJob->caller;
		if (esl_send_recv(handle, ss.str().c_str()) != ESL_SUCCESS)
		{
			//disconnect or fs failed, retry job
			apr_queue_push (jobqueue, pJob);
			return;
		}
		else
		{
			pJob->result = "1";
		}
	}else{
			
			//a-leg
 			string aleg_uuid ;
			string bleg_uuid ;
			
			string getVar = "api create_uuid";	
			esl_send_recv(handle, (getVar).c_str());
			if (handle->last_sr_event != NULL)
			{
				//strcpy(aleg_uuid,handle->last_sr_event->body);
				aleg_uuid = handle->last_sr_event->body;
				LOG4CXX_INFO(gLogger,"aleg_uuid:" << aleg_uuid);
			}
			getVar = "api create_uuid";	
			esl_send_recv(handle, (getVar).c_str());
			if (handle->last_sr_event != NULL)
			{
				bleg_uuid = handle->last_sr_event->body;
				//strcpy(bleg_uuid,handle->last_sr_event->body);
				LOG4CXX_INFO(gLogger,"bleg_uuid:" << bleg_uuid);
			}
			int n = 0;
			ss << "bgapi originate ";
		for (vector<call_param_t>::iterator it = callParams.begin(); it != callParams.end(); ++it)
		{
			if (n > 0)
			{
				ss << "|";
			}
			ss << "{origination_uuid="<< aleg_uuid <<",ignore_early_media=true,proxy_media=true,jitterbuffer_msec=180,call_timeout=50,orig_acct=" << pJob->acct << ",orig_caller=" << pJob->caller << ",orig_called=" << pJob->called << "}";
			ss << "[calltype=1,execute_on_answer=\'sched_hangup +" << it->nTimeout << " alloted_timeout\',origination_caller_id_name=\'" << it->caller << "\',origination_caller_id_number=" << it->caller << ",absolute_codec_string=\'" << it->codec << "\']"
			<< strProtocol << it->called << "@" << it->server;
			n++;
		}
		string uuids = aleg_uuid + "_" + bleg_uuid;
		ss << " A" << bleg_uuid  << " XML ablegring " << pJob->caller << " " << pJob->caller;
		
		LOG4CXX_INFO(gLogger,"aleg:" << ss.str());
		//b-leg
		//检测第二线时长
			if (!getCallParam(pJob->acct, pJob->caller, pJob->called, CALLBACK2, callParamsb))
			{
				//db failed, retry job
				//apr_queue_push (jobqueue, pJob);
				LOG4CXX_INFO(gLogger,"outcall2 getCallParam error:"<< pJob->acct << " caller:" << pJob->caller << " called:" << pJob->called);
				return;
			}

			if (callParamsb.begin()->nTimeout <= 0)
			{
				LOG4CXX_INFO(gLogger,"outcall2 nTimeout error:"<< pJob->acct << " caller:" << pJob->caller << " called:" << pJob->called << " time:" << callParamsb.begin()->nTimeout);
				return;
			}
		stringstream ssb;
		ssb << "api bg_system php \/fs.php \"originate^";
		n = 0;

		for (vector<call_param_t>::iterator it = callParamsb.begin(); it != callParamsb.end(); ++it)
		{
			if (n > 0)
			{
				ssb << "|";
			}
			ssb << "{origination_uuid="<< bleg_uuid <<",ignore_early_media=true,proxy_media=true,jitterbuffer_msec=180,call_timeout=50,orig_acct=" << pJob->acct << ",orig_caller=" << pJob->caller << ",orig_called=" << pJob->called << "}";
			ssb << "[calltype=2,execute_on_answer=\'sched_hangup +" << it->nTimeout << " alloted_timeout\',origination_caller_id_name=\'" << it->caller << "\',origination_caller_id_number=" << it->caller << ",absolute_codec_string=\'" << it->codec << "\']"
			<< strProtocol << it->called << "@" << it->server;
			n++;
		}

		ssb << "^" << "B"<< aleg_uuid << "^XML^ablegring^" << pJob->caller << "^" << pJob->caller << "\"";
		LOG4CXX_INFO(gLogger,"bleg:" << ssb.str());
		
		///////
		if(esl_send_recv(handle, ss.str().c_str()) != ESL_SUCCESS)
		{
			apr_queue_push (jobqueue, pJob);
			return;
		}
		
		if (esl_send_recv(handle, ssb.str().c_str()) != ESL_SUCCESS  )
		{
			//disconnect or fs failed, retry job
			apr_queue_push (jobqueue, pJob);
			return;
		}
		
		//sleep(1000);
		//string uuidstr = "bgapi sched_api +15 none uuid_bridge ";
		//LOG4CXX_INFO(gLogger,"uuid_bridge :" << (uuidstr + " " + aleg_uuid + " " + bleg_uuid + " ").c_str());
		//if (esl_send_recv(handle, (uuidstr + " " + aleg_uuid + " " + bleg_uuid + " ").c_str()) != ESL_SUCCESS)
		//{
			//disconnect or fs failed, retry job
		//	apr_queue_push (jobqueue, pJob);
		//	return;
		//}else{
			pJob->result = "1";
		//}
			
		
	}
    

    apr_queue_push (resultqueue, pJob);
}
void outAuthcall(esl_handle_t* handle, orderAuth_info_t* pJob)
{
	string strProtocol;
	strProtocol = "sofia/external/";

	vector<call_param_t> callParams;

	if (!getCallParam(pJob->acct, pJob->caller, pJob->caller, AUTHCALL, callParams))
	{
		//db failed, retry job
		//apr_queue_push (jobqueue, pJob);
		return;
	}

	if (callParams.begin()->nTimeout <= 0)
	{
		return;
	}

	//设置拨号字符串
	stringstream ss;

	ss << "bgapi originate {ignore_early_media=true,proxy_media=true,jitterbuffer_msec=180,call_timeout=50,orig_acct=" << pJob->acct << ",orig_caller=" << pJob->caller << ",orig_called=" << pJob->authcode << "}";
	int n = 0;

	for (vector<call_param_t>::iterator it = callParams.begin(); it != callParams.end(); ++it)
	{
		if (n > 0)
		{
			ss << "|";
		}

		ss << "[calltype=7,execute_on_answer=\'sched_hangup +" << it->nTimeout << " alloted_timeout\',origination_caller_id_name=\'" << it->caller << "\',origination_caller_id_number=" << it->caller << ",absolute_codec_string=\'" << it->codec << "\']"
			<< strProtocol << it->called << "@" << it->server;
		n++;
	}

	ss << " " << pJob->authcode << " XML auth " << pJob->caller << " " << pJob->caller;

	if (esl_send_recv(handle, ss.str().c_str()) != ESL_SUCCESS)
	{
		//disconnect or fs failed, retry job
		apr_queue_push (authqueue, pJob);
		return;
	}
	else
	{
		pJob->result = "1";
	}
	apr_queue_push (authresultqueue, pJob);
}
static void * APR_THREAD_FUNC auth_thread_run(apr_thread_t *thd, void *obj)
{
    callserver_info_t*   pCtrl = (callserver_info_t*) obj;

    esl_handle_t handle = {{0}};
_reconnect:

    if (esl_connect(&handle, pCtrl->server, pCtrl->port, pCtrl->user, pCtrl->pwd) == ESL_SUCCESS)
    {
        LOG4CXX_INFO(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " success");
    }
    else
    {
        LOG4CXX_WARN(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " failed");
    }

    while (handle.connected)
    {
        orderAuth_info_t* pJob = NULL;

        while (handle.connected && (apr_queue_trypop(authqueue, (void**)&pJob) == APR_SUCCESS))
        {
            outAuthcall(&handle, pJob);
            pJob = NULL;
        }

        if (esl_recv_event_timed(&handle, 100, 1, NULL) == ESL_FAIL)
        {
            break;
        }
    }

    apr_sleep(APR_USEC_PER_SEC);
	goto _reconnect;
    return NULL;
}
static void *msg_thread_run(esl_thread_t *me, void *obj)
{

    callserver_info_t*   pCtrl = (callserver_info_t*) obj;

    esl_handle_t handle = {{0}};

    if (esl_connect(&handle, pCtrl->server, pCtrl->port, pCtrl->user, pCtrl->pwd) == ESL_SUCCESS)
    {
        LOG4CXX_INFO(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " success");
    }
    else
    {
        LOG4CXX_WARN(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " failed");
        goto done;
    }

    while (handle.connected)
    {
        ordercall_info_t* pJob = NULL;

        while (handle.connected && (apr_queue_trypop(jobqueue, (void**)&pJob) == APR_SUCCESS))
        {
            outcall(&handle, pJob);
            pJob = NULL;
        }

        if (esl_recv_event_timed(&handle, 100, 1, NULL) == ESL_FAIL)
        {
            break;
        }
    }

done:
    apr_sleep(APR_USEC_PER_SEC);
    apr_queue_push(conn_queue, pCtrl);
    return NULL;
}





static void sighandler( int sig_no )
{
    gRunThread = 0;
}



#ifndef _DEBUG
int localMain(int argc, char* argv[]);
int main(int argc, char *argv[])
{
    if (init_daemon(argc, argv, localMain) < 0)
    {
        return 0;
    }

    return 0;
}
int localMain(int argc, char* argv[])
#else
int main(int argc, char *argv[])
#endif
{
    signal(SIGUSR1, sighandler);
    apr_initialize();
    CIniFile config;
    string strCfgPath("/etc/fsorder.ini");

    if (argc >= 2)
    {
        strCfgPath = argv[1];
    }

    ostringstream ss;

    config.SetPath(strCfgPath);
    config.ReadFile();
    PropertyConfigurator::configure(strCfgPath);

    const string password = config.GetValue("order", "fsstr", "66call");
    const string strClient = config.GetValue("order", "callservers", "");

    const string strAuthServer = config.GetValue("order", "authservers", "");

    const int    nThreadPool = config.GetValueI("order", "threads", 100);
    const int    nQueuePool = config.GetValueI("order", "jobqueue", 1000);


    apr_thread_pool_t *thrp;


    //initalize线程池
    apr_pool_t *main_Pool;

    if (apr_pool_create(&main_Pool, NULL) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init memory pool ok");

    if (apr_queue_create(&jobqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    if (apr_queue_create(&resultqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_terminate();
        return -1;
    }

    if (apr_queue_create(&conn_queue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init call job queue pool ok");


    if (apr_queue_create(&authqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init call job queue pool ok");


    if (apr_queue_create(&authresultqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init call job queue pool ok");



    if (apr_thread_pool_create(&thrp, nThreadPool, nThreadPool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_thread_pool_create error,thread pool=" << nThreadPool);
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_queue_term(conn_queue);
        apr_pool_destroy(main_Pool);

        apr_terminate();
        return -2;//线程池创建失败
    }

    if (!initDBS(config, "order-db", main_Pool))
    {
        LOG4CXX_WARN(gLogger, "init dbs error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_queue_term(conn_queue);
        apr_pool_destroy(main_Pool);
        apr_terminate();
        return -3;//数据库初始化失败
    }

    //创建客户监视线程
    if (strClient.length() > 0)
    {
        string delimiter(",");
        const vector<string>& clientlist = split(strClient, delimiter);

        for (vector<string>::const_iterator iter = clientlist.begin(); iter != clientlist.end(); ++iter)
        {
            char* pIp = strdup((*iter).c_str());
            char* pPort = strchr(pIp, ':');

            if (pPort != NULL)
            {
                *pPort++ = '\0';
            }
            else
            {
                pPort = "8021";
            }

            callserver_info_t* pCtrl = new callserver_info_t();
            pCtrl->server = strdup(pIp);
            pCtrl->port = atoi(pPort);
            pCtrl->user = NULL;
            pCtrl->pwd = strdup(password.c_str());
            apr_queue_push(conn_queue, pCtrl);
            free(pIp);
        }
    }

    esl_global_set_logger(NULL);
    //创建数据库监视线程
    apr_thread_pool_push(thrp, fetch_order, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);
    apr_thread_pool_push(thrp, result_order, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);

    apr_thread_pool_push(thrp, fetch_auth, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);
    apr_thread_pool_push(thrp, result_auth, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);


    callserver_info_t* pAuthCtrl = new callserver_info_t();
    {
		char* pIp = strdup(strAuthServer.c_str());
        char* pPort = strchr(pIp, ':');

        if (pPort != NULL)
        {
            *pPort++ = '\0';
        }
        else
        {
            pPort = "8021";
        }


        pAuthCtrl->server = strdup(pIp);
        pAuthCtrl->port = atoi(pPort);
        pAuthCtrl->user = NULL;
        pAuthCtrl->pwd = strdup(password.c_str());

        free(pIp);
    }

    apr_thread_pool_push(thrp, auth_thread_run, pAuthCtrl, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);
    


    while (gRunThread)
    {
        callserver_info_t* pCtrl = NULL;
        apr_status_t ret = apr_queue_pop(conn_queue, (void**) & pCtrl);

        if (ret == APR_SUCCESS)
        {
            if (esl_thread_create_detached(msg_thread_run, pCtrl) != ESL_SUCCESS)
            {
                apr_queue_push(conn_queue, pCtrl);
            }
        }
        else if (ret == APR_EOF)
        {
            break;
        }
    }

    gRunThread = 0;
    apr_queue_term(jobqueue);
    apr_queue_term(resultqueue);
    apr_queue_term(authqueue);
    apr_queue_term(authresultqueue);

    apr_queue_term(conn_queue);
    apr_thread_pool_destroy(thrp);

    destoryDBS();
    apr_pool_destroy(main_Pool);
    apr_terminate();
    return 0;
}
